package com.lili.collection.core.runner.pipe;

import cn.hutool.core.lang.Singleton;
import com.lili.collection.core.constans.ExecutorConstants;
import com.lili.collection.core.context.Context;
import com.lili.collection.core.context.PipeContext;
import com.lili.collection.core.context.impl.BlockingQueueExchangeStageContext;
import com.lili.collection.core.context.impl.GenericPipeContext;
import com.lili.collection.core.domain.RunnerMsgTransfer;
import com.lili.collection.core.event.StageEvent;
import com.lili.collection.core.runner.StageRunner;
import com.lili.collection.core.runner.handler.StageContextChangedHandlerFactory;
import com.lili.collection.core.runner.result.RunnerResult;
import lombok.Getter;
import lombok.extern.slf4j.Slf4j;

import java.util.HashMap;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.Future;
import java.util.concurrent.ScheduledThreadPoolExecutor;

/**
 * 通用管道Runner
 *
 * @author lili
 * @date 2023/3/7 11:34
 */
@Slf4j
public class AsyncPipeRunner extends AbstractPipeRunner {

    @Getter
    private PipeContext pipeContext;

    @Getter
    private final ScheduledThreadPoolExecutor runnerExecutor;

    protected Map<String,Future<RunnerResult>> submitMap;

    private boolean runFlag=false;

    public AsyncPipeRunner(Context context) {
        this.pipeContext = new GenericPipeContext(context, new BlockingQueueExchangeStageContext());
        this.runnerExecutor = Singleton.get("runnerExecutor", () -> new ScheduledThreadPoolExecutor(ExecutorConstants.RUNNER_THREAD_POOL_SIZE_DEFAULT));
        this.submitMap = new ConcurrentHashMap<>();
        ensureStageRunner();
    }

    @Override
    public void run() {
        if(runFlag){
            return;
        }
        runFlag=true;
        if (this.stageRunnerList.size() == 0) return;

        this.execPreHandle();
        RunnerMsgTransfer transfer=new RunnerMsgTransfer();
        transfer.setDeleteOffsetNum(this.getOutputRunner().getDeleteOffsetNum());
        for (StageRunner stageRunner : this.stageRunnerList) {
            stageRunner.setRunnerTransfer(transfer);
            Future<RunnerResult> submit = this.runnerExecutor.submit(stageRunner,stageRunner.getRunnerResult());
            submitMap.put(stageRunner.getClass().getName(),submit);
        }
    }

    @Override
    public synchronized void onStageEventTrigger(StageEvent stageEvent, StageRunner.StageType stageType, Object params) {
        StageContextChangedHandlerFactory.createHandler(stageType).handle(this, stageEvent, params);
    }

    @Override
    public Map<String, RunnerResult> getPipeRunnerResult() {
        if(!runFlag){
            this.run();
        }
        Map<String, RunnerResult> resultMap= new HashMap<>();
        for (String runnerName : submitMap.keySet()) {
            try {
                resultMap.put(runnerName,submitMap.get(runnerName).get());
            } catch (Exception e) {
                log.error("获取runner运行结果异常!",e);
            }
        }
        return resultMap;
    }
}